package com.szy.system.common.util.multithreading;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.Collection;
import java.util.concurrent.CountDownLatch;

/**
 * @author : zhenyun.su
 * @comment :线程池来处理任务
 * @since : 2020/2/26
 */

@Deprecated
public class MultithreadingAgent {
    private static Logger logger = LoggerFactory.getLogger(MultithreadingAgent.class);

    private String jobName;
    private Integer threadNumber;
    private ThreadPoolTaskExecutor threadPool;

    public MultithreadingAgent(String jobName, Integer threadNumber) {
        this.jobName = jobName;
        this.threadNumber = threadNumber;
        threadPool= new ThreadPoolTaskExecutor();
        threadPool.setCorePoolSize(threadNumber);
        threadPool.setMaxPoolSize(threadNumber);
//        threadPool.setQueueCapacity(threadNumber);
        threadPool.setThreadNamePrefix(String.format(this.jobName+"-multithreading-"));
    }

    public static MultithreadingAgent getInstance(String jobName, Integer threadNumber) {
        threadNumber = threadNumber >= 1 ? threadNumber : 1;
        logger.debug("getInstance, jobName: {} ; threadNumber: {}", jobName, threadNumber);

        return new MultithreadingAgent(jobName, threadNumber);
    }

    /**
     * 自动使用多线程，把一个集合中的每一个元素提交到线程池中，并执行给定的操作。
     *
     * @param collection 目标集合
     * @param invoker    给定操作
     * @param <T>        元素类型
     */
    public <T extends Object> MultithreadingResult<T> execute(Collection<T> collection, final MultithreadingInvoker<T> invoker) {
        long startTime=System.currentTimeMillis();

        MultithreadingResult<T> multithreadingResult = new MultithreadingResult<>();
        if (collection == null || collection.size() <= 0) {
            multithreadingResult.success("collection is null");
            logger.debug("{} execute, com.daphne.szy.demo.data is null", jobName);
            return multithreadingResult;
        } else {
            multithreadingResult.process(String.format("%s execute, Started processing %d records ", jobName, collection.size()));
            logger.debug(multithreadingResult.getInfo());
        }

        try {
            threadPool.initialize();
            //定义一个反向计数器，数值用集合元素数量。
            CountDownLatch countDownLatch = new CountDownLatch(collection.size());
            for (T item : collection) {
                //提交线程
                threadPool.submit(new Runnable() {
                    @Override
                    public void run() {
                        try{
                            invoker.invoke(item);
                        } catch (Exception e) {
                            logger.debug("{} execute, {} exception: {}: ", jobName, item, e.getMessage());
                        } finally {
                            //反向计数
                            logger.debug("{} execute, A record has been processed ", jobName);
                            countDownLatch.countDown();
                        }
                    }
                });
            }
            //等待反向计数完成
            countDownLatch.await();
            multithreadingResult.success(String.format("%s execute, A total of %d records have been processed successfully ", jobName, collection.size()));
            logger.debug(multithreadingResult.getInfo());
        } catch (Exception e) {
            multithreadingResult.exception(String.format("%s execute, exception: %s ", jobName, e.getMessage()));
            logger.debug(multithreadingResult.getInfo());
        } finally {
            long elapse = System.currentTimeMillis() - startTime;
            multithreadingResult.finish(String.format("finish, consumed : %s millisecond ", elapse));
            logger.debug(multithreadingResult.getInfo());
            threadPool.shutdown();
            return multithreadingResult;
        }
    }
}
